// Copyright 2018 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: develop dispatcher
// feature: modify log level
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: support batch send
// ------------------------------------------------------------------

#include "UDPChannelResource.h"

#include <asio.hpp>
#include "UDPTransportInterface.h"
#include <transport/ReceiverResource.h>
#ifdef EDDS_XDP_TRANSPORT_COMPILE
#include "UDPv4Transport.h"
#endif

#if defined(__unix__)
#include <arpa/inet.h>
#endif

namespace vbs {
namespace transport {

#ifdef EDDS_XDP_TRANSPORT_COMPILE
UDPChannelResource::UDPChannelResource(UDPTransportInterface* transport, vbsUDPSocket& socket, uint32_t maxMsgSize,
                                       const Locator& locator, const std::string& sInterface,
                                       std::shared_ptr<ReceiverResource> receiver, LixdpSocket* lixdp_socket)
#else
UDPChannelResource::UDPChannelResource(UDPTransportInterface* transport, vbsUDPSocket& socket, uint32_t maxMsgSize,
                                       const Locator& locator, const std::string& sInterface,
                                       std::shared_ptr<ReceiverResource> receiver)
#endif
    : ChannelResource(maxMsgSize),
      message_receiver_(std::move(receiver)),
      socket_(std::move(socket)),
#ifdef EDDS_XDP_TRANSPORT_COMPILE
      lixdp_socket_(lixdp_socket),
      socket_type_(SocketType::SOCKET_TYPE_UDP),
#endif
      only_multicast_purpose_(false),
      interface_(sInterface),
      transport_(transport)

{

#ifdef EDDS_XDP_TRANSPORT_COMPILE
    if (lixdp_socket_ && transport_->get_socket_type() == SocketType::SOCKET_TYPE_XDP) {
        std::string thread_name("EVBSRecv");

        logInfo(TRANSPORT, "UDPChannelResource  CREATE locator " << locator);
        (dynamic_cast<LixdpSocketUDP*>(lixdp_socket_))->set_channel_resource(this);
        lixdp_socket_->enable();
        socket_type_ = SocketType::SOCKET_TYPE_XDP;
    } else
#endif
    {
#if defined(__unix__)
        std::stringstream ss;
        // 收包线程名格式为R + ip + port. 其中ip为16进制, 固定占8个字节. port为10进制
        ss << "R" << std::hex << std::setw(8) << std::setfill('0') << ntohl(inet_addr(sInterface.c_str()))
           << std::to_string(socket_.local_endpoint().port());
        std::string thread_name(ss.str());
#else
        std::string thread_name("EVBSRecv");
#endif
        logInfo(TRANSPORT, "UDPChannelResource  CREATE locator " << locator);
        thread(std::thread(&UDPChannelResource::perform_listen_operation, this, locator, thread_name));
    }
}

UDPChannelResource::~UDPChannelResource() {
    message_receiver_ = nullptr;

#ifdef EDDS_XDP_TRANSPORT_COMPILE
    if (lixdp_socket_ && socket_type_ == SocketType::SOCKET_TYPE_XDP) {
        lixdp_socket_->set_alive(false);
        lixdp_socket_->disable();
        delete lixdp_socket_;
        lixdp_socket_ = nullptr;
    } else
#endif
    {
        asio::error_code ec;
        (void)socket()->close(ec);
    }
}

void UDPChannelResource::perform_listen_operation(Locator input_locator, const std::string& thread_name) {
    Locator remote_locator;
    (void)thread_name;
#if defined(__unix__)
    (void)pthread_setname_np(pthread_self(), thread_name.c_str());
#endif  // defined(__unix__)

    while (alive()) {
        // Blocking receive.
        unsigned char* buffer = message_buffer();
        uint32_t length = 0;
        if (!Receive(buffer, buffer_size(), length, remote_locator)) {
            continue;
        }

        // Processes the data through the CDR Message interface.
        if (message_receiver() != nullptr) {
            message_receiver()->OnDataReceived(buffer, length, input_locator, remote_locator);
        } else if (alive()) {
            logInfo(RTPS_MSG_IN, "Received Message, but no receiver attached");
        }
    }

    message_receiver(nullptr);
}

bool UDPChannelResource::Receive(unsigned char* receive_buffer, uint32_t receive_buffer_capacity,
                                 uint32_t& receive_buffer_size, Locator& remote_locator) {
    try {
        asio::ip::udp::endpoint senderEndpoint;

        size_t bytes = socket()->receive_from(asio::buffer(receive_buffer, receive_buffer_capacity), senderEndpoint);
        receive_buffer_size = static_cast<uint32_t>(bytes);
        if (receive_buffer_size > 0) {
            // This is not necessary anymore but it's left here for back compatibility with versions
            // older than 1.8.1
            if ((receive_buffer_size == 13) && (memcmp(receive_buffer, "EPRORTPSCLOSE", 13) == 0)) {
                return false;
            }
            transport_->endpoint_to_locator(senderEndpoint, remote_locator);
        }
        return (receive_buffer_size > 0);
    } catch (const std::exception& error) {
        (void)error;
        logInfo(RTPS_MSG_OUT,
                "Error receiving data: " << error.what() << " - " << message_receiver() << " (" << this << ")");
        return false;
    }
}

void UDPChannelResource::release() {
#ifdef EDDS_XDP_TRANSPORT_COMPILE
    if (socket_type_ == SocketType::SOCKET_TYPE_XDP) {
        // 关闭lixdp接收
        if (lixdp_socket_) {
            lixdp_socket_->set_alive(false);
            lixdp_socket_->disable();
        }
    } else
#endif
    {
        // Cancel all asynchronous operations associated with the socket.
        socket()->cancel();
        // Disable receives on the socket.
        // shutdown always returns a 'shutdown: Transport endpoint is not connected' error,
        // since the endpoint is indeed not connected. However, it unblocks the synchronous receive
        // in Windows and Linux anyways, which is what we want.
        asio::error_code ec;
        (void)socket()->shutdown(asio::socket_base::shutdown_type::shutdown_receive, ec);

#if defined(__APPLE__)
        // On OSX shutdown does not seem to unblock the listening thread, but close does.
        socket()->close();
#endif  // if defined(__APPLE__)
    }
}

inline void UDPChannelResource::clear() {
#ifdef EDDS_XDP_TRANSPORT_COMPILE
    if (socket_type_ == SocketType::SOCKET_TYPE_XDP) {
        alive_.store(false);
        if (lixdp_socket_) {
            lixdp_socket_->disable();
        }
    } else
#endif
    {
        ChannelResource::clear();
    }
}

}  // namespace transport
}  // namespace vbs
